package org.example;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;

public class Test2MapFunction implements MapFunction<Tuple2<Integer, String>, String> {

    private static final Logger LOG = LoggerFactory.getLogger(Test2MapFunction.class);

    @Override
    public String map(Tuple2<Integer, String> jobJson) throws Exception {
        QTreeMapAggre.PathFindJobBase job = QTreeMapAggre.jobFromJson(jobJson.f1);
        long timeMillis = System.currentTimeMillis();
        ArrayList<Point> rs = new ArrayList<>();
        if (job.getClass() == QTreeMapAggre.MultiPointsInOneMap.class) {
            QTreeMapAggre.MultiPointsInOneMap s = (QTreeMapAggre.MultiPointsInOneMap) job;
            if (s.points.size() < 2) {
                return "[]";
            }
            var seri = new QTreeSerializer();
            String fp = "/var/flinkrc/pytile/" + s.mapBase.mx + "_" + s.mapBase.my + ".qt";
            var quadTree = seri.deserializeCustom(fp);
            var afg = new QTreeFromGrid();
            afg.computeAllNeighbors(quadTree);
            //
            ArrayList<QTreeMapAggre.Offset[]> v = new ArrayList<>();
            for (int i = 0; i != s.points.size() - 1; i++) {
                var prev = new QTreeMapAggre.Offset((s.points).get(i).ox, (s.points).get(i).oy);
                var next = new QTreeMapAggre.Offset((s.points).get(i + 1).ox, (s.points).get(i + 1).oy);
                QTreeMapAggre.Offset[] pair = new QTreeMapAggre.Offset[]{prev, next};
                v.add(pair);
            }
            v.forEach(w -> {
                var route = new AStar(quadTree).findPath(new Point(w[0].ox, w[0].oy), new Point(w[1].ox, w[1].oy));
                if (route.isEmpty()) {
                    LOG.error("An empty path was found, from ({},{}) to ({},{})", w[0].ox, w[0].oy, w[1].ox, w[1].oy);
                    LOG.error("A direct line was made to prevent empty route.");
                    var u = getLinePointsDDA(
                            s.mapBase.mx + w[0].ox, s.mapBase.my + w[0].oy,
                            s.mapBase.mx + w[1].ox, s.mapBase.my + w[1].oy);
                    rs.addAll(u);
                } else {
                    ArrayList<Point> route2 = new ArrayList<>();
                    route.forEach(a -> {
                        route2.add(new Point(s.mapBase.mx + a.x, s.mapBase.my + a.y));
                    });
                    rs.addAll(route2);
                }
            });
        } else {
            QTreeMapAggre.TwoPointsInTwoMap s = (QTreeMapAggre.TwoPointsInTwoMap) job;
            var map1 = s.mapBase1;
            var map2 = s.mapBase2;
            var pt1 = s.offset1;
            var pt2 = s.offset2;
            var u = getLinePointsDDA(map1.mx + pt1.ox, map1.my + pt1.oy, map2.mx + pt2.ox, map2.my + pt2.oy);
            rs.addAll(u);
        }
        LOG.info("[****]job {} mainly work is finished in {} ms![****]", jobJson.f0, System.currentTimeMillis() - timeMillis);
        var v = new JSONObject();
        var r = new JSONArray();
        for (var it : rs) {
            var u = new JSONArray();
            u.add(it.x);
            u.add(it.y);
            r.add(u);
        }
        v.put("id", jobJson.f0);
        v.put("points", r);
        return v.toJSONString();
    }

    public static ArrayList<Point> getLinePointsDDA(int x0, int y0, int x1, int y1) {
        ArrayList<Point> points = new ArrayList<>();

        int dx = x1 - x0;
        int dy = y1 - y0;
        int steps = Math.max(Math.abs(dx), Math.abs(dy));

        if (steps == 0) {
            points.add(new Point(x0, y0));
            return points;
        }

        double xIncrement = (double) dx / steps;
        double yIncrement = (double) dy / steps;

        double x = x0;
        double y = y0;

        for (int i = 0; i <= steps; i++) {
            points.add(new Point((int) Math.round(x), (int) Math.round(y)));
            x += xIncrement;
            y += yIncrement;
        }
        return points;
    }

    public static void main(String[] args) throws Exception {
        var g = "hello, main";
        System.out.println(g);
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        var mergedMapPoints = QTreeMapAggre.getJobMapAndPoints();

        ArrayList<Tuple2<Integer, String>> s = new ArrayList<>();
        for (int i = 0; i != mergedMapPoints.size(); i++) {
            s.add(Tuple2.of(i, mergedMapPoints.get(i).toJson()));
        }

        DataStreamSource<Tuple2<Integer, String>> r = env.fromCollection(s);
        var x = r.map(new Test2MapFunction());
        x.print();
        //x.writeAsText("/tmp/flink.out");
        // 创建FileSink
        FileSink<String> sink = FileSink
                .forRowFormat(new Path("./output/path"), new SimpleStringEncoder<String>("UTF-8"))
                .build();
        x.sinkTo(sink);
        env.execute("Flink Java API Skeleton");
    }
}
